package com.hy.study.outputfromat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * @author huyong
 * @date 2024 年 02 月 06 日 10:09: 52
 */
public class FRecordWriter extends RecordWriter<Text, NullWritable> {
    private TaskAttemptContext job;
    private FSDataOutputStream atguigu;
    private FSDataOutputStream other;

    public FRecordWriter(TaskAttemptContext job) {
        this.job = job;
        // 获取文件信息
        try {
            Configuration configuration = job.getConfiguration();
            FileSystem fileSystem = FileSystem.get(configuration);
            // 创建两个文件的输出流
            atguigu = fileSystem.create(new Path("F:\\study\\Hadoop\\output\\fileoutput\\atguigu.log"));
            other = fileSystem.create(new Path("F:\\study\\Hadoop\\output\\fileoutput\\other.log"));
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    /**
     * 真正的写出功能
     *
     * @param key   the key to write.
     * @param value the value to write.
     *
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public void write(Text key, NullWritable value) throws IOException, InterruptedException {
        // 业务逻辑写出
        String keyStr = key.toString();
        if (keyStr.contains("atguigu")) {
            atguigu.write(keyStr.getBytes(StandardCharsets.UTF_8));
        } else {
            other.write(keyStr.getBytes(StandardCharsets.UTF_8));
        }
    }

    /**
     * 关闭资源
     *
     * @param context the context of the task
     *
     * @throws IOException
     * @throws InterruptedException
     */

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        IOUtils.closeStream(atguigu);
        IOUtils.closeStream(other);
    }
}
